package com.atbeijing.app

import java.text.SimpleDateFormat
import java.util

import com.alibaba.fastjson.JSON
import com.atbeijing.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atbeijing.constants.GmallConstants
import com.atbeijing.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.json4s.DefaultFormats
import org.json4s.jackson.Serialization
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer
import collection.JavaConverters._

/**
 *拼接数据存到ES
 */
object SaleDetailApp {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SaleDetailApp")
    val ssc = new StreamingContext(conf,Seconds(5))

    //订单流
    val orderInfoRecordDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER,ssc)
    //订单详情流
    val orderDetailRecordDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL,ssc)

    //转换样例类
    val orderInfoDS: DStream[(String, OrderInfo)] = orderInfoRecordDS.mapPartitions(partition => {
      partition.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
        //补全时间
        val createTime: Array[String] = orderInfo.create_time.split(" ")
        orderInfo.create_date = createTime(0)
        orderInfo.create_hour = createTime(1)
        (orderInfo.id, orderInfo)
      })
    })

    //转换样例类
    val orderDetailDS: DStream[(String,OrderDetail)] = orderDetailRecordDS.mapPartitions(partation => {
      partation.map(record => {
        val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
        (orderDetail.order_id,orderDetail)
      })
    })

    //双流fullouterjoin,避免数据丢失
    val fullJoinDS: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDS.fullOuterJoin(orderDetailDS)



    //关联OrderInfo和OrderDetail
    val infoJoinDetailDS: DStream[SaleDetail] = fullJoinDS.mapPartitions(partition => {
      //可变集合存放OrderInfo关联OrderDetail后的数据,随后返回用于关联userInfo
      val saleDetails = new ListBuffer[SaleDetail]
      //redis连接
      val jedis: Jedis = new Jedis("hadoop202", 6379)

      //没有返回值所以不用map遍历每条记录,只需要将数据放入创建的saleDetails集合
      partition.foreach { case (orderId, (orderInfoOpt, orderDetailOpt)) => {
        //redis key 存放订单 value为string
        val orderInfoKey: String = "orderInfo" + orderId
        //redis key 存放订单详情 value为set
        val orderDetailKey: String = "orderDetail" + orderId
        //样例类转json时要用
        implicit val formats: DefaultFormats.type = org.json4s.DefaultFormats

        //orderInfo不为null
        if (orderInfoOpt.isDefined) {
          //订单信息
          val orderInfo: OrderInfo = orderInfoOpt.get

          //当订单详情不为null时将信息存入集合
          if (orderDetailOpt.isDefined) {
            val orderDetail: OrderDetail = orderDetailOpt.get
            //+= 表示向集合saleDetails添加数据
            saleDetails += new SaleDetail(orderInfo, orderDetail)
          }

          //由于订单和详情是1对多的关系,所以订单都要存到缓存redis,以供后面的详情流匹配
          //样例类转json
          val orderInfoStr: String = Serialization.write(orderInfo)
          //写入redis并设置过期时间
          jedis.set(orderInfoKey, orderInfoStr)
          jedis.expire(orderInfoKey, 20)

          //查询缓存redis中是否有orderDetail,存在就整合放入集合
          //redis key  "orderDetail"+订单id  一个订单有多个detail所以存detail的redis数据类型为集合
          val orderDetailsFromRedis: util.Set[String] = jedis.smembers(orderDetailKey)
          for (elem <- orderDetailsFromRedis.asScala) {
            //查询到数据就整合
            val orderDetail: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
            saleDetails += new SaleDetail(orderInfo, orderDetail)
          }
        } else if (orderDetailOpt.isDefined) { //orderInfo为null且订单详情不为null
          val orderDetail: OrderDetail = orderDetailOpt.get
          //查询缓存中是否有订单数据可以关联,有的话存入集合,没有存入redis
          val str: String = jedis.get(orderInfoKey)
          if (str != null && !"".equals(str)) {
            val orderInfo: OrderInfo = JSON.parseObject(str, classOf[OrderInfo])
            saleDetails += new SaleDetail(orderInfo, orderDetail)
          } else {
            //样例类转json
            val orderDetailStr: String = Serialization.write(orderDetail)
            //写入redis并设置过期时间
            jedis.set(orderDetailKey, orderDetailStr)
            jedis.expire(orderInfoKey, 20)
          }
        }
      }}
      //关闭redis练级
      jedis.close()
      //返回关联后的数据集合
      saleDetails.iterator
    })



    //遍历infoJoinDetailDS从redis拿出userInfo拼接
    val saleDetailDS: DStream[SaleDetail] = infoJoinDetailDS.mapPartitions(partition => {
      val jedis: Jedis = new Jedis("hadoop202", 6379)
      val saleDetails: Iterator[SaleDetail] = partition.map(saleDetail => {
        val userInfoKey: String = "userInfo" + saleDetail.user_id
        val userInfoStr: String = jedis.get(userInfoKey)
        val userInfo: UserInfo = JSON.parseObject(userInfoStr, classOf[UserInfo])
        saleDetail.mergeUserInfo(userInfo)
        saleDetail
      })
      jedis.close()
      saleDetails
    })


    //将关联好的SaleDetail对象写入ES
    val sdf = new SimpleDateFormat("yyyy-MM-dd")
    saleDetailDS.foreachRDD(rdd =>{
      rdd.foreachPartition(partition =>{
        //索引名
        val indexName: String = GmallConstants.ES_SALE_DETAIL_INDEX+sdf.format(System.currentTimeMillis())
        val saleDetailList: Iterator[(String, SaleDetail)] = partition.map(saleDetail => {
          //(文档id,数据)
          (saleDetail.order_id, saleDetail)
        })
        MyEsUtil.insertBulk(indexName,saleDetailList.toList)
      })
    })



    //启动程序
    ssc.start()
    ssc.awaitTermination()
  }
}
